package net.bwie.realtime.jtp.dws.douyin.log.functions;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.dws.douyin.log.bean.DouYinClass;
import org.apache.flink.api.common.functions.MapFunction;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.Instant;
import java.time.format.DateTimeFormatter;

public class DouyinBeanMapFunction implements MapFunction<String, DouYinClass> {
    @Override
    public DouYinClass map(String s) throws Exception {
        JSONObject jsonObject = JSON.parseObject(s);
        String liveRoomId = jsonObject.getString("live_room_id");
        String anchorId = jsonObject.getString("anchor_id");
        String sourceType = jsonObject.getString("source_type");
        String skuId = jsonObject.getString("sku_id");
        double productPrice = jsonObject.getDoubleValue("product_price");
        String productName = jsonObject.getString("product_name");
        String platform = jsonObject.getString("platform");
        int liveDurationSec = jsonObject.getInteger("live_duration_sec");
        String eventType = jsonObject.getString("event_type");
        String imgUrl = jsonObject.getString("img_url");
        String sourceId = jsonObject.getString("source_id");
        String region = jsonObject.getString("region");
        String userId = jsonObject.getString("user_id");
        String payStatus = jsonObject.getJSONObject("event_properties").getString("pay_status");
        double amount = jsonObject.getJSONObject("event_properties").getDoubleValue("amount");
        int orderQty = jsonObject.getJSONObject("event_properties").getIntValue("order_qty");
        String eventTime = jsonObject.getString("event_time");
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        Instant instant = LocalDateTime.parse(eventTime, formatter)
                .atZone(ZoneId.systemDefault())
                .toInstant();
        long timestamp = instant.toEpochMilli();

        return new DouYinClass(null,
                null,
                liveRoomId,
                anchorId,
                sourceType,
                skuId,
                productPrice,
                productName,
                platform,
                liveDurationSec,
                eventType,
                imgUrl,
                sourceId,
                region,
                userId,
                payStatus,
                amount,
                orderQty,
                timestamp
                );
    }

    @Override
    protected Object clone() throws CloneNotSupportedException {
        return super.clone();
    }
}
